package top.wangjiu.reactive.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import top.wangjiu.reactive.rpc.config.ClientConfiguration;
import top.wangjiu.reactive.rpc.decode.RpcResponseDecoder;
import top.wangjiu.reactive.rpc.encode.RpcRequestEncoder;
import top.wangjiu.reactive.rpc.model.RpcRequest;
import top.wangjiu.reactive.rpc.model.RpcResponse;
import top.wangjiu.reactive.rpc.server.RpcServer;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;


public class RpcClient {

    private static final Log log = LogFactory.getLog(RpcServer.class);

    private final InetSocketAddress serverAddress;

    private Channel channel;

    private EventLoopGroup eventLoopGroup;

    private final Bootstrap bootstrap;

    private final Map<String, FluxSink<RpcResponse>> map = new ConcurrentHashMap<>();

    private final Executor executor;

    private final ClientConfiguration configuration;

    private volatile int reconnectTime = 0;

    RpcClient(ClientConfiguration configuration, InetSocketAddress serverAddress){
        this.executor = configuration.getResponseExecutor();
        this.configuration = configuration;
        this.serverAddress = serverAddress;
        eventLoopGroup = new NioEventLoopGroup(new DefaultThreadFactory(getClass(),true));
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new RpcResponseDecoder());
                        ch.pipeline().addLast(new RpcRequestEncoder());
                        ch.pipeline().addLast(new ClientChannel(map,executor,configuration));
                    }
                });
        configuration.getOptions().forEach(bootstrap::option);
        configuration.getAttrs().forEach(bootstrap::attr);
    }

    public boolean isActive(){
        return channel!=null&&channel.isActive();
    }

    public int getProcessingRequestCount(){
        return map.keySet().size();
    }

    public synchronized boolean connect(){
        if(isActive()){
            return true;
        }
        try {
            ChannelFuture future = bootstrap.connect(this.serverAddress).sync();
            channel = future.channel();
            channel.closeFuture().addListener(f->{
                if(configuration.isAutoReconnect() && reconnectTime<configuration.getMaxAutoReconnectTime()){
                    log.info("rpc client reconnect " + reconnectTime + "time to " +this.serverAddress);
                    connect();
                }else{
                    log.info("rpc client reconnect to " +this.serverAddress + " failed!");
                }
            });
            reconnectTime = 0;
        } catch (Exception e) {
            log.error("rpc client connect "+ this.serverAddress + " error",e);
            reconnectTime++;
            return false;
        }
        return true;
    }

    private void writeRequest(RpcRequest request){
        getChannel().writeAndFlush(request);
    }

    private Channel getChannel(){
        if(isActive()){
            return channel;
        }else{
            synchronized (this){
                if(connect()){
                    return channel;
                }else {
                    throw new RuntimeException("connect to server failed!");
                }
            }
        }
    }

    public Flux<RpcResponse> send(RpcRequest request){
        return Flux.create(sink -> {
            map.put(request.getRequestId(),sink);
            sink.onRequest(r->{
                request.setCount(r);
                writeRequest(request);
            });
            sink.onCancel(()-> {
                if(map.containsKey(request.getRequestId())){
                    writeRequest(new RpcRequest(request.getRequestId(),true));
                }
            });
        });
    }

}
